From 357c0868b7ec4ce4a14f1ad9919dc9fbf10fe51e Mon Sep 17 00:00:00 2001 From: vyzo Date: Tue, 18 May 2021 12:20:19 +0300 Subject: [PATCH] proper config for termination batching and commit wait --- extern/storage-sealing/commit_batch.go | 13 ++++--- extern/storage-sealing/sealiface/config.go | 5 +++ extern/storage-sealing/sealing.go | 2 +- extern/storage-sealing/terminate_batch.go | 41 +++++++++++----------- node/config/def.go | 10 ++++-- 5 files changed, 39 insertions(+), 32 deletions(-) diff --git a/extern/storage-sealing/commit_batch.go b/extern/storage-sealing/commit_batch.go index ad0b00e88..6b3cef6d5 100644 --- a/extern/storage-sealing/commit_batch.go +++ b/extern/storage-sealing/commit_batch.go @@ -22,12 +22,6 @@ import ( "github.com/filecoin-project/lotus/extern/sector-storage/ffiwrapper" ) -var ( - // TODO: config! - - CommitBatchWait = 5 * time.Minute -) - const arp = abi.RegisteredAggregationProof_SnarkPackV1 type CommitBatcherApi interface { @@ -87,6 +81,11 @@ func (b *CommitBatcher) run() { var forceRes chan *cid.Cid var lastMsg *cid.Cid + cfg, err := b.getConfig() + if err != nil { + panic(err) + } + for { if forceRes != nil { forceRes <- lastMsg @@ -101,7 +100,7 @@ func (b *CommitBatcher) run() { return case <-b.notify: sendAboveMax = true - case <-time.After(TerminateBatchWait): + case <-time.After(cfg.CommitBatchWait): sendAboveMin = true case fr := <-b.force: // user triggered forceRes = fr diff --git a/extern/storage-sealing/sealiface/config.go b/extern/storage-sealing/sealiface/config.go index f62911b70..1c0945db2 100644 --- a/extern/storage-sealing/sealiface/config.go +++ b/extern/storage-sealing/sealiface/config.go @@ -21,4 +21,9 @@ type Config struct { AggregateCommits bool MinCommitBatch int MaxCommitBatch int + CommitBatchWait time.Duration + + TerminateBatchMax uint64 + TerminateBatchMin uint64 + TerminateBatchWait time.Duration } diff --git a/extern/storage-sealing/sealing.go b/extern/storage-sealing/sealing.go index d990cb02f..35ff15f51 100644 --- a/extern/storage-sealing/sealing.go +++ b/extern/storage-sealing/sealing.go @@ -152,7 +152,7 @@ func New(api SealingAPI, fc FeeConfig, events Events, maddr address.Address, ds notifee: notifee, addrSel: as, - terminator: NewTerminationBatcher(context.TODO(), maddr, api, as, fc), + terminator: NewTerminationBatcher(context.TODO(), maddr, api, as, fc, gc), commiter: NewCommitBatcher(context.TODO(), maddr, api, as, fc, gc, verif), getConfig: gc, diff --git a/extern/storage-sealing/terminate_batch.go b/extern/storage-sealing/terminate_batch.go index 0e96e8384..3833109a1 100644 --- a/extern/storage-sealing/terminate_batch.go +++ b/extern/storage-sealing/terminate_batch.go @@ -21,14 +21,6 @@ import ( "github.com/filecoin-project/lotus/chain/actors/builtin/miner" ) -var ( - // TODO: config - - TerminateBatchMax uint64 = 100 // adjust based on real-world gas numbers, actors limit at 10k - TerminateBatchMin uint64 = 1 - TerminateBatchWait = 5 * time.Minute -) - type TerminateBatcherApi interface { StateSectorPartition(ctx context.Context, maddr address.Address, sectorNumber abi.SectorNumber, tok TipSetToken) (*SectorLocation, error) SendMsg(ctx context.Context, from, to address.Address, method abi.MethodNum, value, maxFee abi.TokenAmount, params []byte) (cid.Cid, error) @@ -38,11 +30,12 @@ type TerminateBatcherApi interface { } type TerminateBatcher struct { - api TerminateBatcherApi - maddr address.Address - mctx context.Context - addrSel AddrSel - feeCfg FeeConfig + api TerminateBatcherApi + maddr address.Address + mctx context.Context + addrSel AddrSel + feeCfg FeeConfig + getConfig GetSealingConfigFunc todo map[SectorLocation]*bitfield.BitField // MinerSectorLocation -> BitField @@ -53,13 +46,14 @@ type TerminateBatcher struct { lk sync.Mutex } -func NewTerminationBatcher(mctx context.Context, maddr address.Address, api TerminateBatcherApi, addrSel AddrSel, feeCfg FeeConfig) *TerminateBatcher { +func NewTerminationBatcher(mctx context.Context, maddr address.Address, api TerminateBatcherApi, addrSel AddrSel, feeCfg FeeConfig, getConfig GetSealingConfigFunc) *TerminateBatcher { b := &TerminateBatcher{ - api: api, - maddr: maddr, - mctx: mctx, - addrSel: addrSel, - feeCfg: feeCfg, + api: api, + maddr: maddr, + mctx: mctx, + addrSel: addrSel, + feeCfg: feeCfg, + getConfig: getConfig, todo: map[SectorLocation]*bitfield.BitField{}, waiting: map[abi.SectorNumber][]chan cid.Cid{}, @@ -113,6 +107,11 @@ func (b *TerminateBatcher) processBatch(notif, after bool) (*cid.Cid, error) { return nil, xerrors.Errorf("getting proving deadline info failed: %w", err) } + cfg, err := b.getConfig() + if err != nil { + return nil, xerrors.Errorf("getting sealing config: %W", err) + } + b.lk.Lock() defer b.lk.Unlock() params := miner2.TerminateSectorsParams{} @@ -193,11 +192,11 @@ func (b *TerminateBatcher) processBatch(notif, after bool) (*cid.Cid, error) { return nil, nil // nothing to do } - if notif && total < TerminateBatchMax { + if notif && total < cfg.TerminateBatchMax { return nil, nil } - if after && total < TerminateBatchMin { + if after && total < cfg.TerminateBatchMin { return nil, nil } diff --git a/node/config/def.go b/node/config/def.go index ba4ec7ce1..6419127d0 100644 --- a/node/config/def.go +++ b/node/config/def.go @@ -242,9 +242,13 @@ func DefaultStorageMiner() *StorageMiner { WaitDealsDelay: Duration(time.Hour * 6), AlwaysKeepUnsealedCopy: true, - AggregateCommits: true, - MinCommitBatch: 1, // we must have at least one proof to aggregate - MaxCommitBatch: 204, // this is the maximum aggregation per FIP13 + AggregateCommits: true, + 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 }, Storage: sectorstorage.SealerConfig{