From 74bad49068956231baf4c48bc69447aa52a95d2c Mon Sep 17 00:00:00 2001 From: vyzo Date: Tue, 18 May 2021 14:30:47 +0300 Subject: [PATCH 1/4] correctly handle commit batch timer --- extern/storage-sealing/commit_batch.go | 88 +++++++++++++++++++++- extern/storage-sealing/sealiface/config.go | 1 + extern/storage-sealing/states_sealing.go | 2 +- node/config/def.go | 7 +- 4 files changed, 90 insertions(+), 8 deletions(-) diff --git a/extern/storage-sealing/commit_batch.go b/extern/storage-sealing/commit_batch.go index 6b3cef6d59b..b61d5cdde93 100644 --- a/extern/storage-sealing/commit_batch.go +++ b/extern/storage-sealing/commit_batch.go @@ -18,6 +18,7 @@ import ( proof5 "github.com/filecoin-project/specs-actors/v5/actors/runtime/proof" "github.com/filecoin-project/lotus/api" + "github.com/filecoin-project/lotus/build" "github.com/filecoin-project/lotus/chain/actors/builtin/miner" "github.com/filecoin-project/lotus/extern/sector-storage/ffiwrapper" ) @@ -27,6 +28,8 @@ const arp = abi.RegisteredAggregationProof_SnarkPackV1 type CommitBatcherApi interface { SendMsg(ctx context.Context, from, to address.Address, method abi.MethodNum, value, maxFee abi.TokenAmount, params []byte) (cid.Cid, error) StateMinerInfo(context.Context, address.Address, TipSetToken) (miner.MinerInfo, error) + StateMarketStorageDeal(context.Context, abi.DealID, TipSetToken) (*api.MarketDeal, error) + ChainHead(ctx context.Context) (TipSetToken, abi.ChainEpoch, error) } type AggregateInput struct { @@ -45,6 +48,7 @@ type CommitBatcher struct { getConfig GetSealingConfigFunc verif ffiwrapper.Verifier + sectors map[abi.SectorNumber]SectorInfo todo map[abi.SectorNumber]AggregateInput waiting map[abi.SectorNumber][]chan cid.Cid @@ -100,7 +104,7 @@ func (b *CommitBatcher) run() { return case <-b.notify: sendAboveMax = true - case <-time.After(cfg.CommitBatchWait): + case <-time.After(b.batchWait(cfg.CommitBatchWait, cfg.CommitBatchSlack)): sendAboveMin = true case fr := <-b.force: // user triggered forceRes = fr @@ -114,6 +118,79 @@ func (b *CommitBatcher) run() { } } +func (b *CommitBatcher) batchWait(maxWait, slack time.Duration) time.Duration { + now := time.Now() + + b.lk.Lock() + defer b.lk.Unlock() + + var deadline time.Time + for sn := range b.todo { + sectorDeadline := b.getSectorDeadline(sn) + if deadline.IsZero() || (!sectorDeadline.IsZero() && sectorDeadline.Before(deadline)) { + deadline = sectorDeadline + } + } + for sn := range b.waiting { + sectorDeadline := b.getSectorDeadline(sn) + if deadline.IsZero() || (!sectorDeadline.IsZero() && sectorDeadline.Before(deadline)) { + deadline = sectorDeadline + } + } + + if deadline.IsZero() { + return maxWait + } + + deadline = deadline.Add(-slack) + if deadline.Before(now) { + return time.Nanosecond // can't return 0 + } + + wait := deadline.Sub(now) + if wait > maxWait { + wait = maxWait + } + + return wait +} + +func (b *CommitBatcher) getSectorDeadline(sn abi.SectorNumber) time.Time { + si, ok := b.sectors[sn] + if !ok { + return time.Time{} + } + + tok, curEpoch, err := b.api.ChainHead(b.mctx) + if err != nil { + log.Errorf("getting chain head: %s", err) + return time.Time{} + } + + deadlineEpoch := si.TicketEpoch + for _, p := range si.Pieces { + if p.DealInfo == nil { + continue + } + + proposal, err := b.api.StateMarketStorageDealProposal(b.mctx, p.DealInfo.DealID, tok) + if err != nil { + log.Errorf("getting deal proposal for %d: %s", p.DealInfo.DealID, err) + continue + } + + if proposal.StartEpoch < deadlineEpoch { + deadlineEpoch = proposal.StartEpoch + } + } + + if deadlineEpoch <= curEpoch { + return time.Now() + } + + return time.Duration(deadlineEpoch-curEpoch) * time.Duration(build.BlockDelaySecs) * time.Second +} + func (b *CommitBatcher) processBatch(notif, after bool) (*cid.Cid, error) { b.lk.Lock() defer b.lk.Unlock() @@ -182,6 +259,7 @@ func (b *CommitBatcher) processBatch(notif, after bool) (*cid.Cid, error) { } delete(b.waiting, sn) delete(b.todo, sn) + delete(b.sectors, sn) return nil }) if err != nil { @@ -192,12 +270,14 @@ func (b *CommitBatcher) processBatch(notif, after bool) (*cid.Cid, error) { } // register commit, wait for batch message, return message CID -func (b *CommitBatcher) AddCommit(ctx context.Context, s abi.SectorNumber, in AggregateInput) (mcid cid.Cid, err error) { +func (b *CommitBatcher) AddCommit(ctx context.Context, s SectorInfo, in AggregateInput) (mcid cid.Cid, err error) { + sn := s.SectorNumber b.lk.Lock() - b.todo[s] = in + b.sectors[sn] = s + b.todo[sn] = in sent := make(chan cid.Cid, 1) - b.waiting[s] = append(b.waiting[s], sent) + b.waiting[sn] = append(b.waiting[sn], sent) select { case b.notify <- struct{}{}: diff --git a/extern/storage-sealing/sealiface/config.go b/extern/storage-sealing/sealiface/config.go index 1c0945db28e..f9784b642e1 100644 --- a/extern/storage-sealing/sealiface/config.go +++ b/extern/storage-sealing/sealiface/config.go @@ -22,6 +22,7 @@ type Config struct { MinCommitBatch int MaxCommitBatch int CommitBatchWait time.Duration + CommitBatchSlack time.Duration TerminateBatchMax uint64 TerminateBatchMin uint64 diff --git a/extern/storage-sealing/states_sealing.go b/extern/storage-sealing/states_sealing.go index 739e8c946fd..6e95fbd6970 100644 --- a/extern/storage-sealing/states_sealing.go +++ b/extern/storage-sealing/states_sealing.go @@ -537,7 +537,7 @@ func (m *Sealing) handleSubmitCommitAggregate(ctx statemachine.Context, sector S return ctx.Send(SectorCommitFailed{xerrors.Errorf("sector had nil commR or commD")}) } - mcid, err := m.commiter.AddCommit(ctx.Context(), sector.SectorNumber, AggregateInput{ + mcid, err := m.commiter.AddCommit(ctx.Context(), sector, AggregateInput{ info: proof.AggregateSealVerifyInfo{ Number: sector.SectorNumber, Randomness: sector.TicketValue, diff --git a/node/config/def.go b/node/config/def.go index 6419127d0ab..553aabb1eee 100644 --- a/node/config/def.go +++ b/node/config/def.go @@ -246,9 +246,10 @@ func DefaultStorageMiner() *StorageMiner { MinCommitBatch: 1, // we must have at least one proof to aggregate MaxCommitBatch: 204, // this is the maximum aggregation per FIP13 CommitBatchWait: time.Day, // this can be up to 6 days - TerminateBatchMin: 1, // same as above - TerminateBatchMax: 204, // same as above - TerminateBatchWait: time.Day, // this can be up to 6 days + CommitBatchSlack: 8 * time.Hour, + TerminateBatchMin: 1, + TerminateBatchMax: 100, + TerminateBatchWait: 5 * time.Minute, }, Storage: sectorstorage.SealerConfig{ From 6b3e04b9b16aeb9679c21331ae6349d93f705cc8 Mon Sep 17 00:00:00 2001 From: vyzo Date: Tue, 18 May 2021 15:28:51 +0300 Subject: [PATCH 2/4] cache sector deadlines. --- extern/storage-sealing/commit_batch.go | 26 +++++++++++--------------- 1 file changed, 11 insertions(+), 15 deletions(-) diff --git a/extern/storage-sealing/commit_batch.go b/extern/storage-sealing/commit_batch.go index b61d5cdde93..b23832fe13b 100644 --- a/extern/storage-sealing/commit_batch.go +++ b/extern/storage-sealing/commit_batch.go @@ -48,9 +48,9 @@ type CommitBatcher struct { getConfig GetSealingConfigFunc verif ffiwrapper.Verifier - sectors map[abi.SectorNumber]SectorInfo - todo map[abi.SectorNumber]AggregateInput - waiting map[abi.SectorNumber][]chan cid.Cid + deadlines map[abi.SectorNumber]time.Time + todo map[abi.SectorNumber]AggregateInput + waiting map[abi.SectorNumber][]chan cid.Cid notify, stop, stopped chan struct{} force chan chan *cid.Cid @@ -67,8 +67,9 @@ func NewCommitBatcher(mctx context.Context, maddr address.Address, api CommitBat getConfig: getConfig, verif: verif, - todo: map[abi.SectorNumber]AggregateInput{}, - waiting: map[abi.SectorNumber][]chan cid.Cid{}, + deadlines: map[abi.SectorNumber]time.Time{}, + todo: map[abi.SectorNumber]AggregateInput{}, + waiting: map[abi.SectorNumber][]chan cid.Cid{}, notify: make(chan struct{}, 1), force: make(chan chan *cid.Cid), @@ -126,13 +127,13 @@ func (b *CommitBatcher) batchWait(maxWait, slack time.Duration) time.Duration { var deadline time.Time for sn := range b.todo { - sectorDeadline := b.getSectorDeadline(sn) + sectorDeadline := b.deadlines[sn] if deadline.IsZero() || (!sectorDeadline.IsZero() && sectorDeadline.Before(deadline)) { deadline = sectorDeadline } } for sn := range b.waiting { - sectorDeadline := b.getSectorDeadline(sn) + sectorDeadline := b.deadlines[sn] if deadline.IsZero() || (!sectorDeadline.IsZero() && sectorDeadline.Before(deadline)) { deadline = sectorDeadline } @@ -155,12 +156,7 @@ func (b *CommitBatcher) batchWait(maxWait, slack time.Duration) time.Duration { return wait } -func (b *CommitBatcher) getSectorDeadline(sn abi.SectorNumber) time.Time { - si, ok := b.sectors[sn] - if !ok { - return time.Time{} - } - +func (b *CommitBatcher) getSectorDeadline(si SectorInfo) time.Time { tok, curEpoch, err := b.api.ChainHead(b.mctx) if err != nil { log.Errorf("getting chain head: %s", err) @@ -259,7 +255,7 @@ func (b *CommitBatcher) processBatch(notif, after bool) (*cid.Cid, error) { } delete(b.waiting, sn) delete(b.todo, sn) - delete(b.sectors, sn) + delete(b.deadlines, sn) return nil }) if err != nil { @@ -273,7 +269,7 @@ func (b *CommitBatcher) processBatch(notif, after bool) (*cid.Cid, error) { func (b *CommitBatcher) AddCommit(ctx context.Context, s SectorInfo, in AggregateInput) (mcid cid.Cid, err error) { sn := s.SectorNumber b.lk.Lock() - b.sectors[sn] = s + b.deadlines[sn] = b.getSectorDeadline(s) b.todo[sn] = in sent := make(chan cid.Cid, 1) From 7512748b56e2e6f03f90297794fdde4e7825ea1c Mon Sep 17 00:00:00 2001 From: vyzo Date: Tue, 18 May 2021 15:31:52 +0300 Subject: [PATCH 3/4] wire in sealing config values --- node/modules/storageminer.go | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/node/modules/storageminer.go b/node/modules/storageminer.go index 1d89a0c4b6b..66052600ebd 100644 --- a/node/modules/storageminer.go +++ b/node/modules/storageminer.go @@ -827,6 +827,11 @@ func NewSetSealConfigFunc(r repo.LockedRepo) (dtypes.SetSealingConfigFunc, error AggregateCommits: cfg.AggregateCommits, MinCommitBatch: cfg.MinCommitBatch, MaxCommitBatch: cfg.MaxCommitBatch, + CommitBatchWait: cfg.CommitBatchWait, + CommitBatchSlack: cfg.CommitBatchSlack, + TerminateBatchMax: cfg.TerminateBatchMax, + TerminateBatchMin: cfg.TerminateBatchMin, + TerminateBatchWait: cfg.TerminateBatchWait, } }) return @@ -845,6 +850,11 @@ func NewGetSealConfigFunc(r repo.LockedRepo) (dtypes.GetSealingConfigFunc, error AggregateCommits: cfg.Sealing.AggregateCommits, MinCommitBatch: cfg.Sealing.MinCommitBatch, MaxCommitBatch: cfg.Sealing.MaxCommitBatch, + CommitBatchWait: cfg.Sealing.CommitBatchWait, + CommitBatchSlack: cfg.Sealing.CommitBatchSlack, + TerminateBatchMax: cfg.Sealing.TerminateBatchMax, + TerminateBatchMin: cfg.Sealing.TerminateBatchMin, + TerminateBatchWait: cfg.Sealing.TerminateBatchWait, } }) return From c544f4ce441f02c89b162c2352d5552ee650a390 Mon Sep 17 00:00:00 2001 From: vyzo Date: Tue, 18 May 2021 16:59:11 +0300 Subject: [PATCH 4/4] avoid extraneous rpc call for storage start epoch --- extern/storage-sealing/commit_batch.go | 12 +++--------- 1 file changed, 3 insertions(+), 9 deletions(-) diff --git a/extern/storage-sealing/commit_batch.go b/extern/storage-sealing/commit_batch.go index b23832fe13b..ba9d988ca03 100644 --- a/extern/storage-sealing/commit_batch.go +++ b/extern/storage-sealing/commit_batch.go @@ -28,7 +28,6 @@ const arp = abi.RegisteredAggregationProof_SnarkPackV1 type CommitBatcherApi interface { SendMsg(ctx context.Context, from, to address.Address, method abi.MethodNum, value, maxFee abi.TokenAmount, params []byte) (cid.Cid, error) StateMinerInfo(context.Context, address.Address, TipSetToken) (miner.MinerInfo, error) - StateMarketStorageDeal(context.Context, abi.DealID, TipSetToken) (*api.MarketDeal, error) ChainHead(ctx context.Context) (TipSetToken, abi.ChainEpoch, error) } @@ -169,14 +168,9 @@ func (b *CommitBatcher) getSectorDeadline(si SectorInfo) time.Time { continue } - proposal, err := b.api.StateMarketStorageDealProposal(b.mctx, p.DealInfo.DealID, tok) - if err != nil { - log.Errorf("getting deal proposal for %d: %s", p.DealInfo.DealID, err) - continue - } - - if proposal.StartEpoch < deadlineEpoch { - deadlineEpoch = proposal.StartEpoch + startEpoch := p.DealInfo.DealSchedule.StartEpoch + if startEpoch < deadlineEpoch { + deadlineEpoch = startEpoch } }