From ba27d452141b6516e23d4a4917dae24af25f96a7 Mon Sep 17 00:00:00 2001 From: Aayush Rajasekaran Date: Tue, 8 Jun 2021 16:46:35 -0400 Subject: [PATCH] Split the getSectorCutoff methods between precommit and commit batchers --- extern/storage-sealing/commit_batch.go | 45 +++++++++++++++++------ extern/storage-sealing/precommit_batch.go | 25 ++++++++++++- 2 files changed, 58 insertions(+), 12 deletions(-) diff --git a/extern/storage-sealing/commit_batch.go b/extern/storage-sealing/commit_batch.go index a31509c08..b88dfe984 100644 --- a/extern/storage-sealing/commit_batch.go +++ b/extern/storage-sealing/commit_batch.go @@ -7,6 +7,10 @@ import ( "sync" "time" + "github.com/filecoin-project/go-state-types/network" + + "github.com/filecoin-project/lotus/chain/actors" + "github.com/ipfs/go-cid" "golang.org/x/xerrors" @@ -34,6 +38,7 @@ type CommitBatcherApi interface { StateSectorPreCommitInfo(ctx context.Context, maddr address.Address, sectorNumber abi.SectorNumber, tok TipSetToken) (*miner.SectorPreCommitOnChainInfo, error) StateMinerInitialPledgeCollateral(context.Context, address.Address, miner.SectorPreCommitInfo, TipSetToken) (big.Int, error) + StateNetworkVersion(ctx context.Context, tok TipSetToken) (network.Version, error) } type AggregateInput struct { @@ -369,16 +374,15 @@ func (b *CommitBatcher) processSingle(mi miner.MinerInfo, sn abi.SectorNumber, i // register commit, wait for batch message, return message CID func (b *CommitBatcher) AddCommit(ctx context.Context, s SectorInfo, in AggregateInput) (res sealiface.CommitBatchRes, err error) { - _, curEpoch, err := b.api.ChainHead(b.mctx) - if err != nil { - log.Errorf("getting chain head: %s", err) - return sealiface.CommitBatchRes{}, nil - } - sn := s.SectorNumber + cu, err := b.getCommitCutoff(s) + if err != nil { + return sealiface.CommitBatchRes{}, err + } + b.lk.Lock() - b.cutoffs[sn] = getSectorCutoff(curEpoch, s) + b.cutoffs[sn] = cu b.todo[sn] = in sent := make(chan sealiface.CommitBatchRes, 1) @@ -452,8 +456,27 @@ func (b *CommitBatcher) Stop(ctx context.Context) error { } } -func getSectorCutoff(curEpoch abi.ChainEpoch, si SectorInfo) time.Time { - cutoffEpoch := si.TicketEpoch + policy.MaxPreCommitRandomnessLookback +func (b *CommitBatcher) getCommitCutoff(si SectorInfo) (time.Time, error) { + tok, curEpoch, err := b.api.ChainHead(b.mctx) + if err != nil { + log.Errorf("getting chain head: %s", err) + return time.Now(), nil + } + + nv, err := b.api.StateNetworkVersion(b.mctx, tok) + if err != nil { + log.Errorf("getting network version: %s", err) + return time.Now(), err + } + + pci, err := b.api.StateSectorPreCommitInfo(b.mctx, b.maddr, si.SectorNumber, tok) + if err != nil { + log.Errorf("getting precommit info: %s", err) + return time.Now(), err + } + + cutoffEpoch := pci.PreCommitEpoch + policy.GetMaxProveCommitDuration(actors.VersionForNetwork(nv), si.SectorType) + for _, p := range si.Pieces { if p.DealInfo == nil { continue @@ -466,10 +489,10 @@ func getSectorCutoff(curEpoch abi.ChainEpoch, si SectorInfo) time.Time { } if cutoffEpoch <= curEpoch { - return time.Now() + return time.Now(), nil } - return time.Now().Add(time.Duration(cutoffEpoch-curEpoch) * time.Duration(build.BlockDelaySecs) * time.Second) + return time.Now().Add(time.Duration(cutoffEpoch-curEpoch) * time.Duration(build.BlockDelaySecs) * time.Second), nil } func (b *CommitBatcher) getSectorCollateral(sn abi.SectorNumber, tok TipSetToken) (abi.TokenAmount, error) { diff --git a/extern/storage-sealing/precommit_batch.go b/extern/storage-sealing/precommit_batch.go index e1ad70911..e1dd42368 100644 --- a/extern/storage-sealing/precommit_batch.go +++ b/extern/storage-sealing/precommit_batch.go @@ -7,6 +7,9 @@ import ( "sync" "time" + "github.com/filecoin-project/lotus/build" + "github.com/filecoin-project/lotus/chain/actors/policy" + "github.com/ipfs/go-cid" "golang.org/x/xerrors" @@ -254,7 +257,7 @@ func (b *PreCommitBatcher) AddPreCommit(ctx context.Context, s SectorInfo, depos sn := s.SectorNumber b.lk.Lock() - b.cutoffs[sn] = getSectorCutoff(curEpoch, s) + b.cutoffs[sn] = getPreCommitCutoff(curEpoch, s) b.todo[sn] = &preCommitEntry{ deposit: deposit, pci: in, @@ -330,3 +333,23 @@ func (b *PreCommitBatcher) Stop(ctx context.Context) error { return ctx.Err() } } + +func getPreCommitCutoff(curEpoch abi.ChainEpoch, si SectorInfo) time.Time { + cutoffEpoch := si.TicketEpoch + policy.MaxPreCommitRandomnessLookback + for _, p := range si.Pieces { + if p.DealInfo == nil { + continue + } + + startEpoch := p.DealInfo.DealSchedule.StartEpoch + if startEpoch < cutoffEpoch { + cutoffEpoch = startEpoch + } + } + + if cutoffEpoch <= curEpoch { + return time.Now() + } + + return time.Now().Add(time.Duration(cutoffEpoch-curEpoch) * time.Duration(build.BlockDelaySecs) * time.Second) +}