more refactoring for interactive porep scheduling
This commit is contained in:
whyrusleeping
+1
-179
@@ -9,11 +9,8 @@ import (
|
||||
logging "github.com/ipfs/go-log"
|
||||
"github.com/libp2p/go-libp2p-core/host"
|
||||
"github.com/pkg/errors"
|
||||
"golang.org/x/xerrors"
|
||||
|
||||
"github.com/filecoin-project/lotus/api"
|
||||
"github.com/filecoin-project/lotus/build"
|
||||
"github.com/filecoin-project/lotus/chain/actors"
|
||||
"github.com/filecoin-project/lotus/chain/address"
|
||||
"github.com/filecoin-project/lotus/chain/events"
|
||||
"github.com/filecoin-project/lotus/chain/store"
|
||||
@@ -87,187 +84,12 @@ func (m *Miner) Run(ctx context.Context) error {
|
||||
|
||||
m.events = events.NewEvents(ctx, m.api)
|
||||
|
||||
go m.handlePostingSealedSectors(ctx)
|
||||
go m.beginPosting(ctx)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *Miner) commitUntrackedSectors(ctx context.Context) error {
|
||||
sealed, err := m.secst.Commited()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
chainSectors, err := m.api.StateMinerSectors(ctx, m.maddr, nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
onchain := map[uint64]struct{}{}
|
||||
for _, chainSector := range chainSectors {
|
||||
onchain[chainSector.SectorID] = struct{}{}
|
||||
}
|
||||
|
||||
for _, s := range sealed {
|
||||
if _, ok := onchain[s.SectorID]; ok {
|
||||
continue
|
||||
}
|
||||
|
||||
log.Warnf("Missing commitment for sector %d, committing sector", s.SectorID)
|
||||
|
||||
if err := m.commitSector(ctx, s); err != nil {
|
||||
log.Error("Committing uncommitted sector failed: ", err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *Miner) handlePostingSealedSectors(ctx context.Context) {
|
||||
incoming := m.secst.Incoming()
|
||||
defer m.secst.CloseIncoming(incoming)
|
||||
|
||||
if err := m.commitUntrackedSectors(ctx); err != nil {
|
||||
log.Error(err)
|
||||
}
|
||||
|
||||
for {
|
||||
select {
|
||||
case sinfo, ok := <-incoming:
|
||||
if !ok {
|
||||
// TODO: set some state variable so that this state can be
|
||||
// visible via some status command
|
||||
log.Warn("sealed sector channel closed, aborting process")
|
||||
return
|
||||
}
|
||||
|
||||
if err := m.commitSector(ctx, sinfo); err != nil {
|
||||
log.Errorf("failed to commit sector: %s", err)
|
||||
continue
|
||||
}
|
||||
|
||||
case <-ctx.Done():
|
||||
log.Warn("exiting seal posting routine")
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Miner) commitSector(ctx context.Context, sinfo sectorbuilder.SectorSealingStatus) error {
|
||||
log.Info("committing sector")
|
||||
|
||||
ssize, err := m.SectorSize(ctx)
|
||||
if err != nil {
|
||||
return xerrors.Errorf("failed to check out own sector size: %w", err)
|
||||
}
|
||||
|
||||
_ = ssize
|
||||
|
||||
// TODO: 2 stage commit
|
||||
/*deals, err := m.secst.DealsForCommit(sinfo.SectorID)
|
||||
if err != nil {
|
||||
return xerrors.Errorf("getting sector deals failed: %w", err)
|
||||
}
|
||||
*/
|
||||
params := &actors.SectorPreCommitInfo{
|
||||
CommD: sinfo.CommD[:],
|
||||
CommR: sinfo.CommR[:],
|
||||
Epoch: sinfo.Ticket.BlockHeight,
|
||||
|
||||
//DealIDs: deals,
|
||||
SectorNumber: sinfo.SectorID,
|
||||
}
|
||||
enc, aerr := actors.SerializeParams(params)
|
||||
if aerr != nil {
|
||||
return errors.Wrap(aerr, "could not serialize commit sector parameters")
|
||||
}
|
||||
|
||||
msg := &types.Message{
|
||||
To: m.maddr,
|
||||
From: m.worker,
|
||||
Method: actors.MAMethods.PreCommitSector,
|
||||
Params: enc,
|
||||
Value: types.NewInt(0), // TODO: need to ensure sufficient collateral
|
||||
GasLimit: types.NewInt(1000000 /* i dont know help */),
|
||||
GasPrice: types.NewInt(1),
|
||||
}
|
||||
|
||||
smsg, err := m.api.MpoolPushMessage(ctx, msg)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "pushing message to mpool")
|
||||
}
|
||||
|
||||
go func() {
|
||||
// TODO: maybe just mark this down in the datastore and handle it differently? This feels complicated to restart
|
||||
mw, err := m.api.StateWaitMsg(ctx, smsg.Cid())
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
randHeight := mw.TipSet.Height() + build.InteractivePoRepDelay
|
||||
|
||||
err = m.events.ChainAt(func(ts *types.TipSet, curH uint64) error {
|
||||
go func() {
|
||||
rand, err := m.api.ChainGetRandomness(ctx, ts, nil, ts.Height()-randHeight)
|
||||
if err != nil {
|
||||
log.Error(errors.Errorf("failed to get randomness for computing seal proof: %w", err))
|
||||
return
|
||||
}
|
||||
|
||||
// TODO: should this get scheduled to preserve proper resource consumption?
|
||||
proof, err := m.secst.SealComputeProof(ctx, sinfo.SectorID, rand)
|
||||
if err != nil {
|
||||
log.Error(errors.Errorf("computing seal proof failed: %w", err))
|
||||
return
|
||||
}
|
||||
|
||||
params := &actors.SectorProveCommitInfo{
|
||||
Proof: proof,
|
||||
SectorID: sinfo.SectorID,
|
||||
//DealIDs: deals,
|
||||
}
|
||||
|
||||
enc, aerr := actors.SerializeParams(params)
|
||||
if aerr != nil {
|
||||
log.Errorf(errors.Wrap(aerr, "could not serialize commit sector parameters"))
|
||||
return
|
||||
}
|
||||
|
||||
msg := &types.Message{
|
||||
To: m.maddr,
|
||||
From: m.worker,
|
||||
Method: actors.MAMethods.ProveCommitSector,
|
||||
Params: enc,
|
||||
Value: types.NewInt(0), // TODO: need to ensure sufficient collateral
|
||||
GasLimit: types.NewInt(1000000 /* i dont know help */),
|
||||
GasPrice: types.NewInt(1),
|
||||
}
|
||||
|
||||
smsg, err := m.api.MpoolPushMessage(ctx, msg)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "pushing message to mpool")
|
||||
}
|
||||
|
||||
// TODO: now wait for this to get included and handle errors?
|
||||
_, err := m.api.StateWaitMsg(ctx, smsg.Cid())
|
||||
if err != nil {
|
||||
log.Errorf("failed to wait for porep inclusion: %s", err)
|
||||
return
|
||||
}
|
||||
|
||||
m.beginPosting(ctx)
|
||||
}()
|
||||
|
||||
return nil
|
||||
}, func(ts *types.TipSet) error {
|
||||
log.Warn("revert in interactive commit sector step")
|
||||
return nil
|
||||
}, 3, mw.TipSet.Height()+build.InteractivePoRepDelay)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
}()
|
||||
|
||||
return nil
|
||||
return m.SealSector(ctx, sinfo.SectorID)
|
||||
}
|
||||
|
||||
func (m *Miner) runPreflightChecks(ctx context.Context) error {
|
||||
|
||||
+10
-34
@@ -31,7 +31,7 @@ type dealMapping struct {
|
||||
Committed bool
|
||||
}
|
||||
|
||||
type TicketFn func(context.Context) (*sectorbuilder.SealSeed, error)
|
||||
type TicketFn func(context.Context) (*sectorbuilder.SealTicket, error)
|
||||
|
||||
// TODO: eventually handle sector storage here instead of in rust-sectorbuilder
|
||||
type Store struct {
|
||||
@@ -60,10 +60,6 @@ func NewStore(sb *sectorbuilder.SectorBuilder, ds dtypes.MetadataDS, tktFn Ticke
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Store) Service() {
|
||||
go s.service()
|
||||
}
|
||||
|
||||
func (s *Store) restartSealing() {
|
||||
sectors, err := s.sb.GetAllStagedSectors()
|
||||
if err != nil {
|
||||
@@ -92,6 +88,15 @@ func (s *Store) restartSealing() {
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Store) SectorStatus(sid uint64) (*sectorbuilder.SectorSealingStatus, error) {
|
||||
status, err := s.sb.SealStatus(sid)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return &status, nil
|
||||
}
|
||||
|
||||
func (s *Store) AddPiece(ref string, size uint64, r io.Reader, dealIDs ...uint64) (sectorID uint64, err error) {
|
||||
sectorID, err = s.sb.AddPiece(ref, size, r)
|
||||
if err != nil {
|
||||
@@ -190,35 +195,6 @@ func (s *Store) SealComputeProof(ctx context.Context, sectorID uint64, rand []by
|
||||
panic("TODO")
|
||||
}
|
||||
|
||||
func (s *Store) CloseIncoming(c <-chan sectorbuilder.SectorSealingStatus) {
|
||||
s.waitingLk.Lock()
|
||||
var at = -1
|
||||
for i, ch := range s.incoming {
|
||||
if ch == c {
|
||||
at = i
|
||||
}
|
||||
}
|
||||
if at == -1 {
|
||||
s.waitingLk.Unlock()
|
||||
return
|
||||
}
|
||||
if len(s.incoming) > 1 {
|
||||
last := len(s.incoming) - 1
|
||||
s.incoming[at] = s.incoming[last]
|
||||
s.incoming[last] = nil
|
||||
}
|
||||
s.incoming = s.incoming[:len(s.incoming)-1]
|
||||
s.waitingLk.Unlock()
|
||||
}
|
||||
|
||||
func (s *Store) Incoming() <-chan sectorbuilder.SectorSealingStatus {
|
||||
ch := make(chan sectorbuilder.SectorSealingStatus, 8)
|
||||
s.waitingLk.Lock()
|
||||
s.incoming = append(s.incoming, ch)
|
||||
s.waitingLk.Unlock()
|
||||
return ch
|
||||
}
|
||||
|
||||
func (s *Store) WaitSeal(ctx context.Context, sector uint64) (sectorbuilder.SectorSealingStatus, error) {
|
||||
s.waitingLk.Lock()
|
||||
watch, ok := s.waiting[sector]
|
||||
|
||||
Reference in New Issue
Block a user